package com.example.mysql.debezium.framework;

import io.debezium.config.Configuration;
import io.debezium.connector.mysql.MySqlConnector;
import org.apache.kafka.connect.storage.FileOffsetBackingStore;

import java.util.UUID;

public class DebeziumConfigBuilder extends Configuration.Builder {

    public static DebeziumConfigBuilder create(String host, int port, String username, String password) {
        DebeziumConfigBuilder builder = new DebeziumConfigBuilder();
        builder.with("database.hostname", host); // IP
        builder.with("database.port", port); // 端口
        builder.with("database.user", username); // 用户
        builder.with("database.password", password); // 密码
        // default
        builder.with("connector.class", MySqlConnector.class.getCanonicalName());
        builder.with("name", "engine");
        builder.with("key.converter.schemas.enable", "false");
        builder.with("value.converter.schemas.enable", "false");
        builder.with("include.schema.changes", "false");
        builder.with("tombstones.on.delete", "false");
        builder.with("database.history.instance.name", UUID.randomUUID().toString());
        builder.with("offset.storage", FileOffsetBackingStore.class.getCanonicalName());
        builder.with("schema.history.internal", io.debezium.storage.file.history.FileSchemaHistory.class.getCanonicalName());
        builder.with("offset.storage.file.filename", "./tmp/offsets.dat");
        builder.with("offset.flush.interval.ms", "6000");
        builder.with("database.server.name", "my_server_01"); // 可以任意修改
        return builder;
    }

    public DebeziumConfigBuilder serverName(String serverName) {
        with("database.server.name", serverName); // 可以任意修改
        return this;
    }

    public DebeziumConfigBuilder tableWhiteList(String database, String table) {
        with("table.whitelist", database + "." + table); // 库.表名
        return this;
    }
}
